package xworker.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.kqueue.KQueueEventLoopGroup;
import io.netty.channel.kqueue.KQueueSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.oio.OioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.channel.socket.oio.OioSocketChannel;
import io.netty.util.concurrent.GenericFutureListener;
import org.xmeta.ActionContext;
import org.xmeta.Thing;
import org.xmeta.util.UtilData;
import xworker.lang.executor.Executor;
import xworker.lang.task.TaskManager;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

public class NettyClient extends NettySession implements NettyClientListener{
	private static final String TAG = NettyClient.class.getName();

	Thing thing;
	ActionContext actionContext;
	ActionContext parentContext;
	private ChannelFuture channelFuture;
	String host;
	int port;
	int waitReconnectTime = 10;
	boolean autoConnect;
	final Object lockObj = new Object();
	final List<NettyClientListener> listeners = new ArrayList<>();
	ScheduledFuture<?> reconnectFuture = null;

	public NettyClient(Thing thing, ActionContext parentContext) {
		super();

		this.thing = thing;
		this.parentContext = parentContext;
		actionContext = new ActionContext();
		actionContext.put("nettyClient", this);
		Map<String, Object> variables = thing.doAction("getVariables", parentContext);
		if(variables != null) {
			actionContext.putAll(variables);
		}
		
		actionContext.put("parentContext", parentContext);
		
		host = thing.doAction("getHost", actionContext);
		port = thing.doAction("getPort", actionContext);

		listeners.add(this);
	}

	public void addListener(NettyClientListener listener){
		if(!listeners.contains(listener)){
			listeners.add(listener);
		}
	}

	public void removeListener(NettyClientListener listener){
		listeners.remove(listener);
	}

	private void checkReconnect(){
		synchronized (lockObj){
			if(autoConnect){
				if((reconnectFuture == null || reconnectFuture.isDone()) && !isConnected()){
					fireBeforeReconnect(waitReconnectTime);
					reconnectFuture = TaskManager.getScheduledExecutorService().schedule(() -> {
						synchronized (lockObj) {
							if(autoConnect && !isConnected()) {
								//先置为null，避免连接失败不能重新启动自动重连
								reconnectFuture = null;
								connect();
							}
						}
					}, waitReconnectTime, TimeUnit.SECONDS);
				}
			}else{
				fireClosed();
			}
		}
	}

	public boolean isAutoConnect() {
		return autoConnect;
	}

	public void setAutoConnect(boolean autoConnect) {
		this.autoConnect = autoConnect;
	}

	public void connect() {
		synchronized (lockObj) {
			if (isConnected()) {
				return;
			}

			int connectTimeOut = thing.doAction("getConnectTimeOut", actionContext);
			String channelType = thing.doAction("getChannelType", actionContext);
			Executor.info(TAG, "Channel type =" + channelType);
			EventLoopGroup workerGroup = null;
			try {
				Bootstrap b = new Bootstrap();

				if("epoll".equals(channelType)){
					workerGroup = new EpollEventLoopGroup();
				}else if("kqueue".equals(channelType)){
					workerGroup = new KQueueEventLoopGroup();
				}else if("oio".equals(channelType)){
					workerGroup = new OioEventLoopGroup();
				}else {
					workerGroup = new NioEventLoopGroup(1);
				}

				b.group(workerGroup);

				for (Thing channels : thing.getAllChilds("Channels")) {
					for (Thing channel : channels.getChilds()) {
						channel.doAction("create", actionContext, "bootstrap", b);
					}
				}

				if("epoll".equals(channelType)){
					b.channelFactory(EpollSocketChannel::new);
				}else if("kqueue".equals(channelType)){
					b.channelFactory(KQueueSocketChannel::new);
				}else if("oio".equals(channelType)){
					b.channelFactory(OioSocketChannel::new);
				}else if(!"other".equals(channelType)){
					b.channelFactory(NioSocketChannel::new);
				}

				b.option(ChannelOption.SO_KEEPALIVE, true);
				b.option(ChannelOption.SO_RCVBUF, 1024 * 1204 * 2);
				b.option(ChannelOption.TCP_NODELAY, true);

				if (connectTimeOut > 0) {
					b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeOut);
				}
				b.handler(new NettyClientChannelInitializer(this));

				// Start the client.
				channelFuture = b.connect(host, port).sync();
				channelFuture.addListener((GenericFutureListener<ChannelFuture>) future -> {
					if (future.isSuccess()) {
						fireConnectSuccess();

						this.fireConnected();
					} else if (future.cause() != null) {
						fireConnectFailure(future.cause());

						checkReconnect();
					} else if (future.isCancelled()) {
						fireConnectCancelled();

						checkReconnect();
					}
				});

				final EventLoopGroup eventLoopGroup = workerGroup;
				channelFuture.channel().closeFuture().addListener((GenericFutureListener<ChannelFuture>) future -> {
					eventLoopGroup.shutdownGracefully();

					fireDisconnected();

					checkReconnect();
				});

				this.setChannel(channelFuture.channel());

				//thing.doAction("startSuccess", actionContext, "nettyClient", NettyClient.this);
			} catch (Exception e) {
				if (workerGroup != null){
					workerGroup.shutdownGracefully();
				}

				fireConnectFailure(e);

				checkReconnect();
			}
		}
	}

	public boolean isConnected() {
		if(isClosed()) {
			return false;
		}
		
		return channelFuture != null && (channelFuture.isDone() && channelFuture.isSuccess());
	}

	public boolean isClosed() {
		if (channelFuture == null) {
			return true;
		}

		if (channelFuture.isDone() && (channelFuture.cause() != null || channelFuture.isCancelled())) {
			return true;
		}

		return channelFuture.channel().closeFuture().isDone();
	}

	public Thing getThing() {
		return thing;
	}

	public ActionContext getActionContext() {
		return actionContext;
	}

	public ChannelFuture getChannelFuture() {
		return channelFuture;
	}

	public int getPort() {
		return port;
	}

	/**
	 * 关闭网络，同时会把自动重连设置为false。
	 */
	public void close() {
		synchronized (lockObj) {
			autoConnect = false;

			if(reconnectFuture != null){
				reconnectFuture.cancel(true);
			}

			if (!isClosed()) {
				channelFuture.channel().close();
			}
		}
	}
	
	public ChannelFuture sendMessage(Object message, boolean flush){
		if(!isClosed()) {
			if(flush){
				return channel.writeAndFlush(message);
			}else{
				return channel.write(message);
			}
		}else{
			throw new ChannelException("Client not connected!");
		}
	}
	
	public ChannelFuture sendMessage(Object message) {
		return sendMessage(message, true);
	}
	
	public void flush() {
		if(!isClosed()) {
			channelFuture.channel().flush();
		}
	}
	
	public void setHost(String host) {
		this.host = host;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public String getHost() {
		return host;
	}

	public int getWaitReconnectTime() {
		return waitReconnectTime;
	}

	public static void onConnectSuccess(ActionContext actionContext) {
		NettyClient client = actionContext.getObject("nettyClient");
		Executor.info(TAG, "Netty client({}) has connected to  {} : {}.", client.getThing().getMetadata().getPath(),				
				client.host, client.getPort());
	}

	public static void onConnectFailure(ActionContext actionContext) {
		NettyClient client = actionContext.getObject("nettyClient");
		Throwable cause = actionContext.getObject("cause");
		Executor.warn(TAG, "Netty client(" + client.getThing().getMetadata().getPath() + ") connect failure", cause);

	}

	public static void onConnectCancelled(ActionContext actionContext) {
		NettyClient client = actionContext.getObject("nettyClient");
		Executor.info(TAG, "Netty client({}) connect cancelled.", client.getThing().getMetadata().getPath());
	}

	public static void onClosed(ActionContext actionContext) {
		NettyClient client = actionContext.getObject("nettyClient");
		Executor.info(TAG, "Netty client({}) is closed.", client.getThing().getMetadata().getPath());
	}

	public static void onDisconnected(ActionContext actionContext){
		NettyClient client = actionContext.getObject("nettyClient");
		Executor.info(TAG, "Netty client({}) is disconnected.", client.getThing().getMetadata().getPath());
	}

	public static void onBeforeReconnect(ActionContext actionContext) {
		NettyClient client = actionContext.getObject("nettyClient");
		Executor.info(TAG, "Netty client({}) is before reconnect, waitReconnectTime=" + client.getWaitReconnectTime(), client.getThing().getMetadata().getPath());
	}

	public static void onConnected(ActionContext actionContext){
		NettyClient client = actionContext.getObject("nettyClient");
		Executor.info(TAG, "Netty client({}) is connected.", client.getThing().getMetadata().getPath());
	}

	public static NettyClient getNettyClient(Thing thing, ActionContext actionContext){
		String key = "nettyClient";
		NettyClient nettyClient;
		if(UtilData.isTrue(thing.doAction("isSingleInstance", actionContext))) {
			nettyClient = thing.getData(key);

			if (nettyClient == null) {
				nettyClient = new NettyClient(thing, actionContext);
				thing.setData(key, nettyClient);
			}
		}else {
			nettyClient = new NettyClient(thing, actionContext);
		}

		return nettyClient;
	}
	public static NettyClient create(ActionContext actionContext) {
		Thing self = actionContext.getObject("self");
		NettyClient nettyClient = getNettyClient(self, actionContext);

		actionContext.g().put(self.getMetadata().getName(), nettyClient);

		return nettyClient;
	}

	public static void remove(ActionContext actionContext) {
		String key = "nettyClient";
		Thing self = actionContext.getObject("self");
		NettyClient nettyClient = self.getData(key);
		if(nettyClient != null) {
			self.setData(key, null);
			nettyClient.close();
		}
	}
	
	public static NettyClient connect(ActionContext actionContext) {
		NettyClient client = create(actionContext);
		client.connect();
		return client;
	}

	public static NettyClient close(ActionContext actionContext) {
		NettyClient client = create(actionContext);
		client.close();
		return client;
	}
	
	public static NettyClient getNettyClient(ActionContext actionContext) {
		return create(actionContext);
	}
	
	public static NettyClient run(ActionContext actionContext) {
		NettyClient client = create(actionContext);
		if(!client.isConnected()) {
			client.connect();
		}
		
		return client;
	}

	public static void doTask(ActionContext actionContext) {
		NettyClient client = actionContext.getObject("nettyClient");
		try {		
			if(client.isClosed()) {
				if(client.waitReconnectTime <= 0){
					client.waitReconnectTime = 10;
					client.connect();
				}
				
				if(client.waitReconnectTime > 0) {
					client.waitReconnectTime --;
				}
				
				if(client.isClosed()) {
					client.getThing().doAction("onReconnect", actionContext);
				}
			}
		}catch(Exception e) {
			Executor.warn(TAG, "Reconnect netty client error, path=" + client.getThing().getMetadata().getPath(), e);
		}
	}

	public void fireConnected(){
		for(NettyClientListener listener : listeners){
			listener.onConnected(this);
		}
	}

	@Override
	public void onConnected(NettyClient nettyClient) {
		thing.doAction("onConnected", actionContext, "nettyClient", nettyClient);
	}

	public void fireDisconnected(){
		for(NettyClientListener listener : listeners){
			listener.onDisconnected(this);
		}
	}

	@Override
	public void onDisconnected(NettyClient nettyClient) {
		thing.doAction("onDisconnected", actionContext, "nettyClient", nettyClient);
	}

	public void fireBeforeReconnect(long waitReconnectTime){
		for(NettyClientListener listener : listeners){
			listener.onBeforeReconnect(this, waitReconnectTime);
		}
	}

	@Override
	public void onBeforeReconnect(NettyClient nettyClient, long waitReconnectTime) {
		thing.doAction("onBeforeReconnect", actionContext, "nettyClient", nettyClient, "waitReconnectTime", waitReconnectTime);
	}

	public void fireClosed(){
		for(NettyClientListener listener : listeners){
			listener.onClosed(this);
		}
	}

	@Override
	public void onClosed(NettyClient nettyClient) {
		thing.doAction("onClosed", actionContext, "nettyClient", nettyClient);
	}

	public void fireConnectSuccess(){
		for(NettyClientListener listener : listeners){
			listener.onConnectSuccess(this);
		}
	}

	@Override
	public void onConnectSuccess(NettyClient nettyClient) {
		thing.doAction("onConnectSuccess", actionContext, "nettyClient", nettyClient);
	}

	public void fireConnectFailure(Throwable cause){
		for(NettyClientListener listener : listeners){
			listener.onConnectFailure(this, cause);
		}
	}

	@Override
	public void onConnectFailure(NettyClient nettyClient, Throwable cause) {
		thing.doAction("onConnectFailure", actionContext, "nettyClient", nettyClient);
	}

	public void fireConnectCancelled(){
		for(NettyClientListener listener : listeners){
			listener.onConnectCancelled(this);
		}
	}

	@Override
	public void onConnectCancelled(NettyClient nettyClient) {
		thing.doAction("onConnectCancelled", actionContext, "nettyClient", nettyClient);
	}
}
